package main

import (
	"errors"
	"fmt"
	"net"
	"time"

	"gitcode.com/shexk/fimqClient/core"
)

type LoginCall func(id uint64, name string, result bool, err string)
type ConnectCall func(result bool, err string)
type ReceiveTopicMsgCall func(msg *core.TopicMessage)
type ReceiveP2PMsgCall func(msg *core.P2PMessage)
type SubsTopicAckCall func(msg *core.SubscriberAck)

type MsgChan struct {
	ID   uint64
	Data []byte
}

func NewMsgChan(id uint64, data []byte) *MsgChan {
	return &MsgChan{ID: id, Data: data}
}

type FimqClient struct {
	tcpclient *core.TcpClient

	OnReceiveTopicMsg ReceiveTopicMsgCall
	OnReceiveP2PMsg   ReceiveP2PMsgCall
	OnLogin           LoginCall
	OnConnect         ConnectCall
	OnSubscriberAck   SubsTopicAckCall
	OnUnSubscriberAck SubsTopicAckCall

	TopicDataChan chan *MsgChan
	P2PDataChan   chan *MsgChan

	ClientID     uint64
	ClientName   string
	TopicAckChan chan uint64
	P2PAckChan   chan uint64

	HeartBeatTimeTick *time.Ticker
	AES128            string
}

func NewFimqClient() *FimqClient {
	return &FimqClient{tcpclient: nil,
		TopicDataChan: make(chan *MsgChan, 1000),
		P2PDataChan:   make(chan *MsgChan, 1000),
		TopicAckChan:  make(chan uint64, 1000),
		P2PAckChan:    make(chan uint64, 1000)}
}
func (c *FimqClient) CloseClient() {
	if c.HeartBeatTimeTick != nil {
		c.HeartBeatTimeTick.Stop()
	}
	close(c.P2PAckChan)
	close(c.TopicAckChan)
	close(c.TopicDataChan)
	close(c.P2PDataChan)
}
func (c *FimqClient) Login(clientid uint64, pwd string) error {
	if c.tcpclient != nil {
		cc := core.NewConnectMsg(clientid, pwd)
		if buf, err := core.EncodeConnectMsg(cc); err != nil {
			return err
		} else {
			return c.tcpclient.SendBytes(buf)
		}
	}
	return errors.New("client not exist")
}

// 具备自动重连功能
func (c *FimqClient) ConnectFimqServer(addr string) {
	c.tcpclient = core.NewTcpClient()
	c.tcpclient.OnClientState = func(conn net.Conn, state int, err error, data []byte, datasize uint32) {
		switch state {
		case core.TCP_STATE_CONNECTED:
			core.GetLog().ShowMsg(core.Log_Debug, "Client Connect Success.")
			if c.OnConnect != nil {
				c.OnConnect(true, "")
			}
		case core.TCP_STATE_DISCONNECTED:
			if c.OnConnect != nil {
				c.OnConnect(false, "Client ReConnect")
			}
			core.GetLog().ShowMsg(core.Log_Debug, "Client ReConnect,Error:", err)
		case core.TCP_STATE_DATA:
			c.ProcessData(data)
		case core.TCP_STATE_ERROR:
			if c.OnConnect != nil {
				c.OnConnect(false, "Client StateError")
			}
			core.GetLog().ShowMsg(core.Log_Debug, "Client StateError,Error:", err)
		}
	}
	c.tcpclient.AutoConnectServer(addr)
}
func (c *FimqClient) StartHeartBeatTimeTick(mills int) {
	go func() {
		c.HeartBeatTimeTick = time.NewTicker(time.Duration(mills) * time.Millisecond)
		defer c.HeartBeatTimeTick.Stop() // Ensure the ticker is stopped when done

		for range c.HeartBeatTimeTick.C {
			data, err := core.EncodeHeartBeatMsg(&core.HeartBeatMsg{
				HeartBeatTime: core.GetNanoID(),
				ClientID:      c.ClientID,
			})
			if err != nil {
				//fmt.Println("Error encoding HeartBeatMsg:", err)
			} else if err := c.tcpclient.SendBytes(data); err != nil {
				//fmt.Println("Error sending HeartBeatMsg:", err)
			}
		}
	}()
}
func (c *FimqClient) ProcessData(data []byte) error {
	switch data[0] {
	case core.MQ_ConnectAck:
		msg, err := core.DecodeConnectAckMsg(data)
		if err != nil {
			fmt.Println("Error decoding ConnectAckMsg:", err)
			return err
		} else {
			if msg.Result == 0 {
				//验证成功，开始执行业务
				c.ClientName = msg.UserName
				c.AES128 = msg.AES128
				core.GetLog().ShowMsg(core.Log_Debug, "Login Success,UserID=", msg.UserID, ",UserName=", msg.UserName, ",AES128=", msg.AES128)
				if c.OnLogin != nil {
					c.OnLogin(msg.UserID, msg.UserName, true, "")
				}
				c.StartHeartBeatTimeTick(1000)
			} else {
				core.GetLog().ShowMsg(core.Log_Error, "Login Error:"+msg.Error.Error())
				if c.OnLogin != nil {
					c.OnLogin(msg.UserID, msg.UserName, false, msg.Error.Error())
				}
			}
		}
	case core.MQ_SubAck:
		msg, err := core.DecodeSubscribeAckMsg(data, true)
		if err != nil {
			fmt.Println("Error SubscriberAck Protocol")
			return err
		}
		if c.OnSubscriberAck != nil {
			c.OnSubscriberAck(msg)
		}
	case core.MQ_UnSubAck:
		msg, err := core.DecodeUnSubscribeAckMsg(data)
		if err != nil {
			fmt.Println("Error SubscriberAck Protocol")
			return err
		}
		if c.OnUnSubscriberAck != nil {
			c.OnUnSubscriberAck(msg)
		}
	case core.MQ_Pub:
		msg, err := core.DecodePublishMessage(data)
		if err != nil {
			fmt.Println("Error decoding PublishMessage:", err)
			return err
		}
		if c.OnReceiveTopicMsg != nil {
			c.OnReceiveTopicMsg(msg)
		}
		if msg.Retain == core.MQ_OnlyOneUse {
			//发送ACK消息
			if buf, err := core.EncodeFinishAckMessage(msg.MessageID, msg.TopicName); err != nil {
				return err
			} else {
				if err := c.tcpclient.SendBytes(buf); err != nil {
					return err
				}
			}
		}
	case core.MQ_PubAck:
		if id, err := core.DecodePublishAckMessage(data); err != nil {
			return err
		} else {
			c.TopicAckChan <- id
		}
	case core.MQ_P2P:
		msg, err := core.DecodeP2PMessage(data)
		if err != nil {
			fmt.Println("Error decoding P2PMessage:", err)
			return err
		}
		if c.OnReceiveP2PMsg != nil {
			c.OnReceiveP2PMsg(msg)
		}
	case core.MQ_P2PAck:
		if id, err := core.DecodeOneByOneAckMessage(data); err != nil {
			return err
		} else {
			c.P2PAckChan <- id
		}
	case core.MQ_DisConnect:
		//你被服务端关闭了，不能再自动重连了
		c.tcpclient.DisableAutoConnect()
	}
	return nil
}

func (c *FimqClient) SubscriberTopic(topics []string) error {
	sub := &core.Subscriber{
		ClientID:  c.ClientID,
		TopicsQos: make(map[string]uint8),
	}
	for t := range topics {
		sub.TopicsQos[topics[t]] = 0
	}

	buf, err := core.EncodeSubscribeMsg(sub)
	if err != nil {
		fmt.Println("Error encoding Subscriber:", err)
		return err
	}
	if c.tcpclient != nil {
		return c.tcpclient.SendBytes(buf) // client.SendBytes(buf)
	} else {
		return fmt.Errorf("client not exist")
	}
}
func (c *FimqClient) UnSubscriberTopic(topics []string) error {
	sub := &core.Subscriber{
		ClientID:  c.ClientID,
		TopicsQos: make(map[string]uint8),
	}
	for t := range topics {
		sub.TopicsQos[topics[t]] = 0
	}

	buf, err := core.EncodeUnSubscribeMsg(sub)
	if err != nil {
		fmt.Println("Error encoding UnSubscriber:", err)
		return err
	}
	if c.tcpclient != nil {
		return c.tcpclient.SendBytes(buf)
	} else {
		return fmt.Errorf("client not exist")
	}
}

func (c *FimqClient) PublishTopicMessage(topic string, data []byte, qos uint8) error {
	pub := core.NewPublishMessage(topic, data, qos, 0x00)
	buf, err := core.EncodePublishMessage(pub)
	if err != nil {
		return err
	}
	switch pub.QoS {
	case core.MQ_QOS_0:
		err = c.tcpclient.SendBytes(buf)
		if err != nil {
			return err
		}
	case core.MQ_QOS_1:
		c.TopicDataChan <- NewMsgChan(pub.MessageID, buf)
	}
	return nil
}
func (c *FimqClient) ProcessMsgChan() {
	go func() {
		for data := range c.P2PDataChan {
			for i := 0; i < core.MQ_ReSendNum; i++ {
				if err := c.tcpclient.SendBytes(data.Data); err != nil {
				}
				id := <-c.P2PAckChan
				if data.ID == id {
					break
				}
			}

		}
	}()
	go func() {
		for data := range c.TopicDataChan {
			for i := 0; i < core.MQ_ReSendNum; i++ {
				if err := c.tcpclient.SendBytes(data.Data); err != nil {
				}
				id := <-c.TopicAckChan
				if data.ID == id {
					break
				}
			}
		}
	}()
}
func (c *FimqClient) PublishP2PMessage(TargetClientID uint64, data []byte, qos uint8) error {
	obj := core.NewP2PMessage(c.ClientID, TargetClientID, data, qos, 0x00)
	buf, err := core.EncodeP2PMessage(obj)
	if err != nil {
		return err
	}
	switch obj.QoS {
	case core.MQ_QOS_0:
		if err = c.tcpclient.SendBytes(buf); err != nil {
			return err
		}
	case core.MQ_QOS_1:
		c.P2PDataChan <- NewMsgChan(obj.MessageID, buf)
	}
	return nil
}
